package com.credithc.signature.callback.configure.rabbitmq;

import com.credithc.signature.callback.mq.core.CoreCallbackConsumer;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.HashMap;
import java.util.Map;

/**
 * 描述：rabbitmq配置<br/>
 * 作者：hujingbo
 * 创建：2019/6/4 14:40
 * 版本：v1.0.0<br/>
 */
@Slf4j
@Configuration
@Data
public class RabbitConfig {

    @Value("${spring.rabbitmq.host}")
    private String host;

    @Value("${spring.rabbitmq.port}")
    private int port;

    @Value("${spring.rabbitmq.username}")
    private String username;

    @Value("${spring.rabbitmq.password}")
    private String password;

    @Value("${spring.rabbitmq.virtual-host:/}")
    private String virtualHost;

    @Value("${spring.rabbitmq.customConfig.exchangeName:signature-callback-exchange}")
    private String exchangeName;

    @Value("${spring.rabbitmq.customConfig.queueName:signature-callback-queue}")
    private String queueName;

    @Value("${spring.rabbitmq.customConfig.vipExchangeName:signature-callback-exchange-vip}")
    private String vipExchangeName;

    @Value("${spring.rabbitmq.customConfig.vipQueueName:signature-callback-queue-vip}")
    private String vipQueueName;

    @Value("${spring.rabbitmq.cache.connection.size:30}")
    private int connectionCacheSize;

    @Value("${spring.rabbitmq.cache.channel.size:25}")
    private int channelCacheSize;

    @Value("${spring.rabbitmq.customConfig.maxConcurrentConsumers:1}")
    private int maxConcurrentConsumers;

    @Value("${spring.rabbitmq.publisher-confirms}")
    private boolean publisherConfirm;

    @Value("${spring.rabbitmq.callback.channel.prefetchCount:10}")
    private int prefetchCount;

    /**
     * 连接工厂
     *
     * @return
     */
    @Bean
    public ConnectionFactory connectionFactory() {
        CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
        connectionFactory.setUsername(username);
        connectionFactory.setPassword(password);
        connectionFactory.setVirtualHost(virtualHost);
        connectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CONNECTION);
        connectionFactory.setChannelCacheSize(channelCacheSize);
        connectionFactory.setConnectionCacheSize(connectionCacheSize);
        connectionFactory.setPublisherConfirms(publisherConfirm);
        connectionFactory.setConnectionLimit(connectionCacheSize);
        return connectionFactory;
    }

    //-----------------------------------------core模块相关----------------------------------------------

    @Bean
    public RabbitAdmin rabbitAdmin() {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory());
        rabbitAdmin.setAutoStartup(true);
        return rabbitAdmin;
    }

    @Bean
    public Queue queue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(queueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;

    }

    @Bean
    public DirectExchange exchange(RabbitAdmin rabbitAdmin) {

        DirectExchange directExchange = new DirectExchange(exchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;

    }

    @Bean
    public Binding binding(RabbitAdmin rabbitAdmin) {
        Queue queue = queue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(exchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }


    @Bean
    public Queue vipQueue(RabbitAdmin rabbitAdmin) {
        Queue queue = new Queue(vipQueueName, true, false, false);
        rabbitAdmin.declareQueue(queue);
        return queue;

    }

    @Bean
    public DirectExchange vipExchange(RabbitAdmin rabbitAdmin) {
        DirectExchange directExchange = new DirectExchange(vipExchangeName, true, false);
        rabbitAdmin.declareExchange(directExchange);
        return directExchange;
    }


    @Bean
    public Binding vipBinding(RabbitAdmin rabbitAdmin) {
        Queue queue = vipQueue(rabbitAdmin);
        Binding binding = BindingBuilder.bind(queue).to(vipExchange(rabbitAdmin)).with(queue.getName());
        rabbitAdmin.declareBinding(binding);
        return binding;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMandatory(true);
        template.setMessageConverter(new Jackson2JsonMessageConverter());
        return template;
    }


    @Bean
    public MessageConverter jsonMessageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public SimpleMessageListenerContainer messageContainer(CoreCallbackConsumer listener) {

        return getSimpleMessageListenerContainer(listener, queue(rabbitAdmin()));
    }

    private SimpleMessageListenerContainer getSimpleMessageListenerContainer(ChannelAwareMessageListener listener, Queue queue) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
        //container.setQueues(coreQueue(rabbitAdmin()));
        container.setQueueNames(queue.getName());
        // 需要将channel暴露给listener才能手动确认，AcknowledgeMode.MANUAL时必须为ture
        container.setExposeListenerChannel(true);
        container.setMaxConcurrentConsumers(maxConcurrentConsumers);
        container.setConcurrentConsumers(maxConcurrentConsumers);
        //设置确认模式手工确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        container.setMessageListener(listener);
        container.setPrefetchCount(prefetchCount);
        // 设为true，扔回queue头部，设为false，丢弃
        container.setDefaultRequeueRejected(true);
        container.setMessageConverter(jsonMessageConverter());
        return container;
    }

    @Bean
    public SimpleMessageListenerContainer vipMessageContainer(CoreCallbackConsumer listener) {
        return getSimpleMessageListenerContainer(listener, vipQueue(rabbitAdmin()));
    }


    //-----------------------------------------延迟队列相关----------------------------------------------

    /**
     * 发送到该队列的message会在一段时间后过期进入到delay_process_queue
     * 每个message可以控制自己的失效时间
     */
    public static String DELAY_QUEUE_NAME;

    @Value("${spring.rabbitmq.delay.queue.name:callback_delay_queue}")
    public void setDelayQueueName(String delayQueueName) {
        DELAY_QUEUE_NAME = delayQueueName;
    }

    /**
     * DLX
     */
    public static String DELAY_EXCHANGE_NAME;

    @Value("${spring.rabbitmq.delay.exchange.name:callback_delay_exchange}")
    public void setDelayExchangeName(String delayExchangeName) {
        DELAY_EXCHANGE_NAME = delayExchangeName;
    }

    /**
     * routing_key
     */
    public static String DELAY_ROUTING_KEY;

    @Value("${spring.rabbitmq.delay.routing.key:callback_delay_key}")
    public void setRoutingKey(String routingKey) {
        DELAY_ROUTING_KEY = routingKey;
    }

    @Bean
    public Queue immediateQueue() {
        return new Queue(DELAY_QUEUE_NAME);
    }

    @Bean
    public CustomExchange customExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-delayed-type", "direct");
        return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true, false, args);
    }

    @Bean
    public Binding bindingNotify(@Qualifier("immediateQueue") Queue queue,
                                 @Qualifier("customExchange") CustomExchange customExchange) {
        return BindingBuilder.bind(queue).to(customExchange).with(DELAY_ROUTING_KEY).noargs();
    }

}
